package cn.com.bluemoon.daps.sync.service.impl;

import cn.com.bluemoon.daps.common.domain.UserInfoHolder;
import cn.com.bluemoon.daps.common.toolkit.DbTypeHelper;
import cn.com.bluemoon.daps.common.enums.DatabaseType;
import cn.com.bluemoon.daps.common.enums.SyncStatus;
import cn.com.bluemoon.daps.common.enums.SyncToEnv;
import cn.com.bluemoon.daps.common.exception.DapThrowException;
import cn.com.bluemoon.daps.common.sql.SqlScriptUtils;
import cn.com.bluemoon.daps.common.toolkit.BmAssetUtils;
import cn.com.bluemoon.daps.model.dto.ModelTableDto;
import cn.com.bluemoon.daps.model.entity.*;
import cn.com.bluemoon.daps.notice.core.NoticeType;
import cn.com.bluemoon.daps.notice.template.DevSyncMessage;
import cn.com.bluemoon.daps.notice.template.ProSyncMessage;
import cn.com.bluemoon.daps.notice.template.base.BaseMessage;
import cn.com.bluemoon.daps.sync.config.PtServerConfig;
import cn.com.bluemoon.daps.sync.core.sync.GlobalConfig;
import cn.com.bluemoon.daps.sync.core.sync.GlobalConfig.TableIdentity;
import cn.com.bluemoon.daps.sync.core.sync.TableModel;
import cn.com.bluemoon.daps.sync.core.sync.log.SyncEnvLog;
import cn.com.bluemoon.daps.sync.core.sync.sql.mysql.MysqlCreateTableSql;
import cn.com.bluemoon.daps.sync.core.sync.sql.mysql.datatype.DataType;
import cn.com.bluemoon.daps.sync.core.sync.sql.mysql.datatype.DataTypeHelper;
import cn.com.bluemoon.daps.sync.entity.DapSyncProdEnvTask;
import cn.com.bluemoon.daps.sync.feignclient.DapMetadataModuleFacade;
import cn.com.bluemoon.daps.sync.feignclient.DapModelModuleFacade;
import cn.com.bluemoon.daps.sync.feignclient.DapNoticeModuleFacade;
import cn.com.bluemoon.daps.sync.feignclient.DapSystemModuleFacade;
import cn.com.bluemoon.daps.sync.service.DapSyncBasicService;
import cn.com.bluemoon.daps.system.dto.SystemUserDto;
import cn.com.bluemoon.daps.system.entity.DapSystemDatasource;
import cn.com.bluemoon.daps.system.entity.DapSystemInfo;
import cn.hutool.core.comparator.CompareUtil;
import cn.hutool.core.util.NumberUtil;
import cn.hutool.db.Db;
import com.alibaba.druid.DbType;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Service;
import reactor.util.function.Tuple2;
import reactor.util.function.Tuple4;
import reactor.util.function.Tuples;

import java.util.*;
import java.util.concurrent.*;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Collectors;

import static cn.hutool.core.text.CharSequenceUtil.trim;

/**
 * 同步结构相关的基础类
 * 1.初始化同步基础信息
 * 2.获取元数据
 * 3.获取模型数据
 * 4.发送邮件
 *
 * @author Jarod.Kong
 */
@Service
@Slf4j
public class DapSyncBasicServiceImpl implements DapSyncBasicService {
    @Autowired
    private DapModelModuleFacade modelModuleFacade;

    @Autowired
    private DapSystemModuleFacade systemModuleFacade;

    @Autowired
    private DapMetadataModuleFacade metadataModuleFacade;

    @Autowired
    private DapNoticeModuleFacade noticeModuleFacade;

    @Autowired
    private PtServerConfig ptServerConfig;

    @Autowired
    private Environment environment;

    /**
     * 环境是否为生产
     *
     * @return t 是 f 否
     */
    private boolean isProdEnv() {
        return Arrays.stream(environment.getActiveProfiles()).anyMatch("prod"::equalsIgnoreCase);
    }

    /**
     * 根据模型id构建同步的全局基础参数
     *
     * @param modelId   模型id
     * @param syncToEnv 同步到
     * @return GlobalConfig
     */
    @Override
    public GlobalConfig buildSyncGlobalConfig(String modelId, SyncToEnv syncToEnv) {
        return buildSyncGlobalConfig(modelId, syncToEnv, UserInfoHolder.getUserInfoOrEmpty());
    }

    /**
     * 根据模型id构建同步的全局基础参数
     * 不尝试连接数据库
     *
     * @param modelId   模型id
     * @param syncToEnv 同步到
     * @param userInfo  执行人-用户信息
     * @return GlobalConfig
     */
    @Override
    public GlobalConfig buildSyncGlobalConfig(String modelId, SyncToEnv syncToEnv, UserInfoHolder.UserInfo userInfo) {
        return buildSyncGlobalConfig(modelId, syncToEnv, userInfo, false);
    }

    /**
     * 根据模型id构建同步的全局基础参数
     *
     * @param modelId      模型id
     * @param syncToEnv    同步到
     * @param userInfo     执行人-用户信息
     * @param tryConnectDb 尝试连接数据库，若失败则爆出异常
     * @return GlobalConfig
     */
    @Override
    public GlobalConfig buildSyncGlobalConfig(String modelId,
                                              SyncToEnv syncToEnv,
                                              UserInfoHolder.UserInfo userInfo,
                                              boolean tryConnectDb) {
        return buildSyncGlobalConfig(modelId, syncToEnv, userInfo, false, null);
    }

    /**
     * 根据模型id构建同步的全局基础参数
     *
     * @param modelId         模型id
     * @param syncToEnv       同步到
     * @param userInfo        执行人-用户信息
     * @param tryConnectDb    尝试连接数据库，若失败则爆出异常
     * @param selectedGroupId 选择导入模型的哪个组中，可为null，null为默认组，非空取指定id（若为空则取默认组）
     * @param logBizId      日志任务id（日志存储id）
     * @return GlobalConfig
     */
    @Override
    public GlobalConfig buildSyncGlobalConfig(String modelId,
                                              SyncToEnv syncToEnv,
                                              UserInfoHolder.UserInfo userInfo,
                                              boolean tryConnectDb,
                                              String selectedGroupId,
                                              String logBizId) {
        DapDataModel model = modelModuleFacade.getByIdAsset(modelId, "模型不存在！！");
        Optional<DapSystemInfo> systemInfo = systemModuleFacade.getSystemInfo(model.getSystemId());
        BmAssetUtils.isTrue(systemInfo.isPresent(), "获取系统信息失败！！");
        // 2021/6/16 根据syncToEnv数据源环境+数据源id获取源数据信息
        GlobalConfig.DataSourceAndEnv dsDto = systemModuleFacade.getSysDataSourceAndEnvByIdAndEnv(model.getDataSourceId(), syncToEnv);
        DapSystemDatasource ds = dsDto.getDs();
        BmAssetUtils.isTrue(DatabaseType.MYSQL.equals(ds.getDatabaseType()), "该功能仅支持MySQL数据库");
        List<DapDataModelGroup> defaultGroups = modelModuleFacade.initModelDefaultGroup(modelId);
        BmAssetUtils.isFalse(defaultGroups.isEmpty(), "模型不存在默认组！！");
        DapDataModelGroup selectedGroup = defaultGroups.get(0);
        if (StringUtils.isNotBlank(selectedGroupId)) {
            selectedGroup = defaultGroups.stream().filter(g -> g.getId().equals(selectedGroupId)).findFirst().orElse(selectedGroup);
        }
        // 增加初始化配置ptosc-server信息
        BmAssetUtils.isTrue(ptServerConfig != null && ptServerConfig.getPtSsh() != null, "ptosc服务器信息不存在！不可执行ptosc命令");
        // global
        GlobalConfig globalConfig = GlobalConfig.builder()
                .logBizKey(logBizId)
                .checkSqlLayer(Boolean.FALSE)
                .model(model)
                .dataSourceAndEnv(dsDto)
                .addSink(false)
                .canDropDbTable(false)
                .modelNewest(true)// 默认取最新的 build中必须指定非默认值，否则会拿默认值false
                .dbType(DbTypeHelper.getDruidDbType(ds.getDatabaseType(), DbType.mysql))
                .system(systemInfo.get())
                .syncToEnv(syncToEnv)
                .userInfoExecutor(userInfo)
                .defaultGroup(selectedGroup)
                .ptServerConfig(ptServerConfig)
                .build();
//        2021/6/16 根据最新数据源设计调整该赋值
        Boolean isSync = globalConfig.getDataSource().getIsSync();
        BmAssetUtils.isTrue(isSync, "模型未开启同步功能！！");
        if (tryConnectDb) {
            log.warn("同步前，验证数据库连接是否可用");
            try {
                log.warn("测试前尝试初始化库名");
                globalConfig.initDbName(globalConfig.isSyncSandboxMock());
                Db db = globalConfig.testConnectDB();
                db.query("select 1");
                log.info("test db pass");
            } catch (Exception e) {
                log.error("数据库连接信息错误", e);
                throw new DapThrowException("模型数据库连接失败，请检查同步环境的连接信息");
            }
        }
        return globalConfig;
    }


    /**
     * 获取模型信息
     *
     * @return list
     */
    @Override
    public List<ModelTableDto> getLocalModelTableDtos(GlobalConfig globalConfig) {
        Tuple4<List<DapDataModelTable>, List<DapDataModelTableField>,
                List<DapDataModelSubordinateKey>, List<DapDataModelTablePartition>>
                modelInfos = modelModuleFacade.queryModelInfosByModelAndNewestAndTableIds(globalConfig.getModel().getId(), globalConfig.isModelNewest(),
                globalConfig.getTableIdentityOpt().map(TableIdentity::getTableIdsDistinct).orElse(null));
        log.info("获取到模型表：{}，字段：{}，索引：{}，分区：{}", modelInfos.getT1().size(), modelInfos.getT2().size(),
                modelInfos.getT3().size(), modelInfos.getT4().size());
        log.info("开始整合表、字段、索引信息");
        Map<String, List<DapDataModelTableField>> tableNameAndFields = modelInfos.getT2().stream().collect(Collectors.groupingBy(DapDataModelTableField::getTableName));
        Map<String, DapDataModelTableField> fieldIdAndField = modelInfos.getT2().stream().collect(Collectors.toMap(DapDataModelTableField::getId, a -> a, (a, b) -> a));
        Map<String, List<DapDataModelSubordinateKey>> tableNameAndIndexes = modelInfos.getT3().stream().collect(Collectors.groupingBy(DapDataModelSubordinateKey::getTableName));
        List<ModelTableDto> collect = modelInfos.getT1().stream().map(t -> {
            List<DapDataModelSubordinateKey> indexes = tableNameAndIndexes.getOrDefault(t.getEngName(), Collections.emptyList());
            indexes.forEach(i -> {
                List<DapDataModelTableField> indexFields = Arrays.stream(i.getFieldIdsSplit())
                        .filter(fieldIdAndField::containsKey).map(fieldIdAndField::get).collect(Collectors.toList());
                i.setIndexFields(indexFields);
                i.setFieldNames(indexFields.stream().map(DapDataModelTableField::getFieldName).collect(Collectors.joining(DapDataModelSubordinateKey.INDEX_FIELDIDS_CHAR)));
            });
            return ModelTableDto.builder().schemaName(t.getTableSchema())
                    .table(t)
                    .fields(tableNameAndFields.getOrDefault(t.getEngName(), Collections.emptyList()).stream().sorted((v1, v2) -> CompareUtil.compare(v1.getFieldRank(), v2.getFieldRank())).collect(Collectors.toList()))
                    .indexes(indexes)
                    .partitions(Collections.emptyList())
                    .build();
        }).collect(Collectors.toList());
        return collect;
    }

    /**
     * @return 获取模型对应的开发库的数据集
     */
    @Override
    public List<TableModel> getMetadataTableModel(GlobalConfig globalConfig, Set<String> tableIds) {
        List<TableModel> tableModels = getMetadataTableModel(globalConfig);
        List<TableModel> metadataTables = tableModels.stream().filter(d -> new ToTableModelPredicate(tableIds).test(d)).collect(Collectors.toList());
        log.info("获取选中的查询元数据的表，{}, size={}", metadataTables, metadataTables.size());
        return metadataTables;
    }

    /**
     * @return 获取模型对应的开发库的数据集
     */
    @Override
    public List<TableModel> getMetadataTableModel(GlobalConfig globalConfig) {
        Tuple2<String, List<ModelTableDto>> dbAndTables = metadataModuleFacade.getMetadataDataModelTableDtos(globalConfig);
        if (dbAndTables == null) {
            return Collections.emptyList();
        }
        List<ModelTableDto> metadataDataModelTableDtos = dbAndTables.getT2();
        String dbNameSchema = dbAndTables.getT1();
        final String finalDbNameSchema = trim(dbNameSchema);
        List<TableModel> modelV2 = metadataDataModelTableDtos.stream().map(new ToTableModelFun(finalDbNameSchema, globalConfig)).collect(Collectors.toList());
        return modelV2;
    }

    /**
     * @return 获取本地模型的数据集
     */
    @Override
    public List<TableModel> getLocalTableModel(GlobalConfig globalConfig) {
        List<ModelTableDto> localModelTableDtos = getLocalModelTableDtos(globalConfig);
        List<TableModel> modelV1 = localModelTableDtos.stream().map(new ToTableModelFun(null, globalConfig)).collect(Collectors.toList());
        return modelV1;
    }

    /**
     * 根据SyncToEnv情况进行过滤表，若SyncToEnv为空则不过滤
     *
     * @param tableNames set
     * @return 获取本地+元数据集+过滤非所选表
     */
    @Override
    public Tuple2<List<TableModel>, List<TableModel>> getLocalAndMetadataTableModel(GlobalConfig globalConfig,
                                                                                    Set<String> tableNames) {
        if (tableNames == null) tableNames = Collections.emptySet();
        Tuple2<List<TableModel>, List<TableModel>> tuple2 = getLocalAndMetadataTableModel(globalConfig);
        List<TableModel> t1 = tuple2.getT1();
        Set<String> finalTableNames = tableNames;
        if (SyncToEnv.DEV.equals(globalConfig.getSyncToEnv()) || SyncToEnv.TEST.equals(globalConfig.getSyncToEnv())) {
            finalTableNames.stream().filter(NumberUtil::isNumber).findFirst().ifPresent(d -> log.warn("开发同步-同步中过滤表，入参d为表名，发现表id为全数据，可能为表id而不是表名, param:{}", d));
            List<TableModel> tbs1 = t1.stream().filter(d -> new ToTableModelPredicate(finalTableNames).test(d)).collect(Collectors.toList());
            List<TableModel> tbs2 = tuple2.getT2().stream().filter(d -> new ToTableModelPredicate(finalTableNames).test(d)).collect(Collectors.toList());
            return Tuples.of(tbs1, tbs2);
        }
        if (SyncToEnv.MODEL.equals(globalConfig.getSyncToEnv())) {
            finalTableNames.stream().filter(NumberUtil::isNumber).findFirst().ifPresent(d -> log.warn("导入开发库表-同步中过滤表，入参d为表名，发现表id为全数据，可能为表id而不是表名, param:{}", d));
            List<TableModel> t2 = tuple2.getT2();
            // 库表到模型没有id，取表名
            List<TableModel> tbs1 = tuple2.getT1().stream().filter(d -> new ToTableModelPredicate(finalTableNames).test(d)).collect(Collectors.toList());
            List<TableModel> tbs2 = t2.stream().filter(d -> new ToTableModelPredicate(finalTableNames).test(d)).collect(Collectors.toList());
            return Tuples.of(tbs1, tbs2);
        }
        return tuple2;
    }

    /**
     * @return 获取本地+元数据集
     */
    @Override
    public Tuple2<List<TableModel>, List<TableModel>> getLocalAndMetadataTableModel(GlobalConfig globalConfig) {
        ExecutorService queryMts = Executors.newFixedThreadPool(2);
        Future<List<TableModel>> localFuture = queryMts.submit(() -> getLocalTableModel(globalConfig));
        Future<List<TableModel>> metaFuture = queryMts.submit(() -> getMetadataTableModel(globalConfig));
        List<TableModel> modelV1, modelV2;
        try {
            try {
                modelV2 = metaFuture.get();
            } catch (InterruptedException | ExecutionException e) {
                throw new DapThrowException("异步获取元数据失败！！", e);
            }
            try {
                modelV1 = localFuture.get();
            } catch (InterruptedException | ExecutionException e) {
                throw new DapThrowException("异步获取模型数据失败！！", e);
            }
        } finally {
            queryMts.shutdown();
            log.info("查询元数据/模型数据结束，关闭线程池");
        }
        if (modelV1 == null) modelV1 = Collections.emptyList();
        if (modelV2 == null) modelV2 = Collections.emptyList();
        return Tuples.of(modelV1, modelV2);
    }

    /**
     * 发送环境邮件
     *
     * @param syncEnvLog 开发同步、发布同步日志
     * @param async      异步
     */
    @Override
    public void sendDevEmail(SyncEnvLog syncEnvLog, GlobalConfig globalConfig, boolean async) {
        if (async) {
            try {
                // 异步中需要调用FeigClient会出现token丢失问题
                log.warn("启动异步发送邮件，可能异步中需要调用FeigClient会出现token丢失问题");
                CompletableFuture.runAsync(() -> sendDevEmail0(syncEnvLog, globalConfig)).get();
            } catch (InterruptedException | ExecutionException e) {
                log.error("异步发送邮件失败！", e);
            }
        } else {
            sendDevEmail0(syncEnvLog, globalConfig);
        }
    }

    /**
     * 不异步发送环境邮件
     *
     * @param syncEnvLog 开发同步、发布同步日志
     */
    @Override
    public void sendDevEmail(SyncEnvLog syncEnvLog) {
        sendDevEmail(syncEnvLog, syncEnvLog.getGlobalConfig(), false);
    }

    /**
     * 不异步发送环境邮件
     *
     * @param syncEnvLog 开发同步、发布同步日志
     */
    @Override
    public void sendDevEmail(SyncEnvLog syncEnvLog, GlobalConfig globalConfig) {
        sendDevEmail(syncEnvLog, globalConfig, false);
    }

    /**
     * 发送环境邮件
     *
     * @param syncEnvLog   开发同步、发布同步日志
     * @param globalConfig g
     */
    private void sendDevEmail0(SyncEnvLog syncEnvLog, GlobalConfig globalConfig) {
        DevSyncMessage message = new DevSyncMessage();
        message.setHeader(String.format("用户%s的%sDDL语句执行结束，相关信息如下", systemModuleFacade.systemUserName(syncEnvLog.getSyncExecBy()), syncEnvLog.getSyncEnv().getBizDesc()));
        message.setModelId(syncEnvLog.getModelId());
        DapDataModel model = syncEnvLog.getGlobalConfig().getModel();
        message.setModelName(model.getName());
        message.setDataSourceName(globalConfig.getDataSource().getName());
        message.setBizDesc(model.getBizDescLine());
        message.setTableId(syncEnvLog.getTableId());
        message.setTableName(syncEnvLog.getTableName());
        message.setSyncEnv(syncEnvLog.getSyncEnv().getDescEnv(isProdEnv()));
        message.setResult(SyncStatus.SYNC_DONE.equals(syncEnvLog.getSyncStatus()) ? BaseMessage.SyncStatus.Success : BaseMessage.SyncStatus.Fail);
        message.setReason(SyncStatus.SYNC_DONE.equals(syncEnvLog.getSyncStatus()) ? null : syncEnvLog.getSyncLog());
        message.setFooter(SyncStatus.SYNC_DONE.equals(syncEnvLog.getSyncStatus()) ? "本邮件仅通知，无需人工处理" :
                String.format("如需处理，请登录数据资产平台：\n     *         https://%s.bluemoon.com.cn/bd-data-asset-platform-view", isProdEnv() ? "domp" : "dompapi"));
        message.setSender(syncEnvLog.getSyncExecBy());
        message.setAction(String.format("模型%s到%s", syncEnvLog.getSyncEnv().getBizDesc(), syncEnvLog.getSyncEnv().getDescEnv()));
        message.setSyncStatus(message.getResult());
        Set<String> emails = systemModuleFacade.getSysDbaRoleUser().stream().map(SystemUserDto::getEmail).collect(Collectors.toSet());
        emails.add(syncEnvLog.getSyncExecByEmail());
        message.setRcs(emails);
        message.setCcs(Lists.newArrayList());
        message.setBcc(Lists.newArrayList());
        message.setTopic(String.format("模型管理-%s：模型同步到%s%s通知-系统消息", syncEnvLog.getSyncEnv().getBizDesc(),
                syncEnvLog.getSyncEnv().getDescEnv(), (SyncStatus.SYNC_DONE.equals(syncEnvLog.getSyncStatus()) ? "成功" : "失败")));
        message.setAttachmentMap(Maps.newHashMap());
        boolean status = noticeModuleFacade.sendDevMsg(NoticeType.EMAIL, message);
        log.info("发送邮件：status:{}", status);
    }

    /**
     * 发送环境邮件
     *
     * @param task         任务单号
     * @param globalConfig 环境类型
     */
    @Override
    public void sendEmailPublishSync(DapSyncProdEnvTask task, GlobalConfig globalConfig, boolean async) {
        if (async) {
            try {
                log.warn("启动异步发送邮件，可能异步中需要调用FeigClient会出现token丢失问题");
                CompletableFuture.runAsync(() -> sendEmailPublishSync0(task, globalConfig)).get();
            } catch (InterruptedException | ExecutionException e) {
                log.error("异步发送邮件失败！", e);
            }
        } else {
            sendEmailPublishSync0(task, globalConfig);
        }
    }


    private void sendEmailPublishSync0(DapSyncProdEnvTask task, GlobalConfig globalConfig) {
        sendEmailPublishSync0(task, globalConfig, false);
    }

    @Override
    public void sendEmailPublishSyncFromSyncManagement(DapSyncProdEnvTask task, GlobalConfig globalConfig) {
        sendEmailPublishSync0(task, globalConfig, true);
    }

    /**
     * 发送
     *
     * @param task         任务
     * @param globalConfig 环境
     */
    private void sendEmailPublishSync0(DapSyncProdEnvTask task, GlobalConfig globalConfig, boolean isSyncManagePass) {
        SyncToEnv syncToEnv = globalConfig.getSyncToEnv();
        BmAssetUtils.isTrue(SyncToEnv.PRE_PROD.equals(syncToEnv) || SyncToEnv.PROD.equals(syncToEnv), "目前允许预发布和生产环境可以发送消息");
        String syncBy = task.getSyncBy(syncToEnv);
        SyncStatus syncStatus = task.getSyncStatus(syncToEnv);
        DapDataModel model = modelModuleFacade.getByIdAsset(task.getModelId(), "");
        ProSyncMessage message = new ProSyncMessage();
        message.setModelId(task.getModelId());
        message.setModelName(model.getName());
        message.setDataSourceName(globalConfig.getDataSource().getName());
        message.setBizDesc(model.getBizDescLine());
        message.setChangeOrder(task.getChangeOrder());
        message.setChangeDesc(task.getChangeDesc());
        message.setSyncToEnv(syncToEnv.getDescEnv(isProdEnv()));
        message.setSyncStatus(SyncStatus.SYNC_DONE.equals(syncStatus) ? BaseMessage.SyncStatus.Success : BaseMessage.SyncStatus.Fail);
        message.setFooter(SyncStatus.SYNC_DONE.equals(syncStatus) ? "本邮件仅通知，无需人工处理" :
                String.format("如需处理，请登录数据资产平台：\n     *         https://%s.bluemoon.com.cn/bd-data-asset-platform-view", isProdEnv() ? "domp" : "dompapi"));
        message.setSender(syncBy);
        message.setAction(String.format("模型%s到%s", syncToEnv.getBizDesc(), syncToEnv.getDescEnv()));
        Set<String> emails = systemModuleFacade.getSysDbaRoleUser().stream().map(SystemUserDto::getEmail).collect(Collectors.toSet());
        emails.add(task.getSyncExecByEmail(syncToEnv));
        message.setRcs(emails);
        log.info("发送给：{}", emails);
        message.setCcs(Lists.newArrayList());
        message.setBcc(Lists.newArrayList());
        message.setAttachmentMap(Maps.newHashMap());
        message.setSyncBy(systemModuleFacade.systemUserName(syncBy));
        if (isSyncManagePass) {
            // 模型管理-模型结构同步：同步状态更改通知-系统消息
            message.setTopic(String.format("模型管理-%s：模型同步到%s-同步状态更改通知-系统消息", syncToEnv.getBizDesc(), syncToEnv.getDescEnv()));
            message.setMoreData(String.format("DBA%s已手工同步该变更内容至%s", globalConfig.getUserInfoExecutor().getUserName(), syncToEnv.getDescEnv()));
        } else {
            message.setTopic(String.format("模型管理-%s：模型同步到%s-DDL执行%s通知-系统消息", syncToEnv.getBizDesc(), syncToEnv.getDescEnv(),
                    (SyncStatus.SYNC_DONE.equals(syncStatus) ? "成功" : "失败")));
            message.setHeader(String.format("用户%s的%sDDL语句执行结束，相关信息如下", systemModuleFacade.systemUserName(syncBy), syncToEnv.getBizDesc()));
        }
        boolean status = noticeModuleFacade.sendProMsg(NoticeType.EMAIL, message);
        log.info("发送邮件：status:{}", status);
    }

    /**
     * 不异步发送环境邮件
     *
     * @param task         任务单号
     * @param globalConfig 环境类型
     */
    @Override
    public void sendEmailPublishSync(DapSyncProdEnvTask task, GlobalConfig globalConfig) {
        sendEmailPublishSync(task, globalConfig, false);
    }

    /**
     * 统一处理本地+元数据结果集
     */
    public static class ToTableModelFun implements Function<ModelTableDto, TableModel> {
        private final String dbNameSchema;
        private final GlobalConfig globalConfig;

        public ToTableModelFun(String dbNameSchema, GlobalConfig globalConfig) {
            this.dbNameSchema = dbNameSchema;
            this.globalConfig = globalConfig;
        }

        @Override
        public TableModel apply(ModelTableDto i) {
            TableModel t = new TableModel();
            i.getTable().setTableSchema(dbNameSchema);
            i.getTable().setDbName(dbNameSchema);
            i.getTable().setBmKey(trim(i.getTable().getEngName()));
            i.getTable().setName(SqlScriptUtils.formatComment(i.getTable().getName()));
            t.setTable(i.getTable());
            i.getFields().forEach(f -> {
                // trim
                f.setBmKey(trim(f.getFieldName()));
                if ("".equals(trim(f.getCompoundType()))) {
                    f.setCompoundType(null);
                }
                f.setComment(SqlScriptUtils.formatComment(f.getComment()));
                f.setLength(trim(f.getLength()));
                f.setDataType(trim(f.getDataType()));
                f.setFieldName(trim(f.getFieldName()));
                // 2021/7/30 模型表定义的空用''表示，而库表的空为空,null还是null
                if ("".equals(f.getDefaultValue())) {
                    f.setDefaultValue("''");
                }
                // 和模型保持统一，元数据保持统一
                DataType dataType = DataTypeHelper.get(f.getDataType());
                if (dataType != null) {
                    Integer scale = dataType.normalizeScale(f.getDataType(), f.getDecimalLength());
                    f.setDecimalLength(scale);
                }
                // 要是该字段为主键，则设置notNull为true
                if (f.getPrimaryKey() != null && f.getPrimaryKey() == 1) {
                    f.setNotNull(1);
                }
                // 兼容mysql5.7、8.0的元数据中整型类型的长度处理
                boolean integerType = DataTypeHelper.isIntegerType(f.getDataType());
                if (integerType) {
                    f.setLength(null);
                    f.setDecimalLength(null);
                }
            });
            t.setFields(i.getFields());
            i.getIndexes().forEach(f -> {
                f.setBmKey(trim(f.getKeyName()));
                f.setFieldIds(trim(f.getFieldIds()));
                f.setFieldNames(trim(f.getFieldNames()));
                f.setKeyChinName(trim(f.getKeyChinName()));
                // 针对存在一份数据中为空的就给为null，统一不提供空
                if (StringUtils.isEmpty(f.getKeyChinName())) {
                    f.setKeyChinName(null);
                }
                if (f.getKeyChinName() != null) {
                    f.setKeyChinName(SqlScriptUtils.formatComment(f.getKeyChinName()));
                }
                // 2021/8/2 修正数据库和模型中索引类型为hash的，都统一为btree
                f.setKeyName(trim(f.getKeyName()));
                if (f.getKeyMethod() != null && 3 == f.getKeyMethod()) {
                    f.setKeyMethod(2);
                    log.warn("索引类型为{}，程序修正为{}，index:{}", 3, 2, f);
                }
            });
            t.setIndexes(i.getIndexes());
            i.getPartitions().forEach(f -> {
                f.setBmKey(trim(f.getPartitionFieldName()));
                f.setPartitionFieldName(trim(f.getPartitionFieldName()));
                f.setPartitionFieldType(f.getPartitionFieldType());
            });
            t.setPartitions(i.getPartitions());
            if (globalConfig != null && DbType.mysql.equals(globalConfig.getDbType())) {
                MysqlCreateTableSql createTableSql = new MysqlCreateTableSql();
                createTableSql.setFields(t.getFields());
                createTableSql.setIndexes(t.getIndexes());
                createTableSql.setTable(t.getTable());
                t.setCreateTableSql(createTableSql.toString());
                t.setCreateTableSqls(createTableSql.createTableSql());
            }
            return t;
        }
    }

    /**
     * 库表过滤器
     */
    public static class ToTableModelPredicate implements Predicate<TableModel> {
        private final Set<String> tableEngNames;

        public ToTableModelPredicate(Set<String> tableEngNames) {
            this.tableEngNames = tableEngNames;
        }


        @Override
        public boolean test(TableModel d) {
            /* 为空全部放行不过滤*/
            if (tableEngNames == null || tableEngNames.isEmpty()) return true;
            return tableEngNames.contains(d.getTable().getEngName());
        }
    }
}
